package com.example.rxbyhand;

import com.example.rxbyhand.scheduler.Scheduler;

import java.util.ArrayDeque;
import java.util.Deque;

/**
 * Author by lp,on 2021/8/17/017,11:44.
 */
public class ObservableObserveOn<T> extends AbstractObservanleWithUpStream<T, T> {
    final Scheduler scheduler;

    public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }


    @Override
    protected void subscribeActual(Observer observer) {
        Scheduler.Worker worker = scheduler.creteWorker();
        source.subscribe(new ObservableObserver<>(observer, worker));

    }

    class ObservableObserver<T> implements Observer<T>, Runnable {

        final Observer<T> downStream;
        final Scheduler.Worker worker;
        final Deque<T> deque;

        volatile boolean done;
        volatile Throwable throwable;
        volatile boolean over;

        ObservableObserver(Observer<T> downStream, Scheduler.Worker worker) {
            this.downStream = downStream;
            this.worker = worker;
            deque = new ArrayDeque<>();
        }

        @Override
        public void onSubscribe() {
            downStream.onSubscribe();

        }

        @Override
        public void onNext(T t) {
            deque.offer(t);
            schedule();
        }

        private void schedule() {
            worker.schedule(this);
        }

        @Override
        public void onComplete() {
        }

        @Override
        public void onError(Throwable throwable) {
            downStream.onError(throwable);
        }

        @Override
        public void run() {
            drainNormal();

        }

        private void drainNormal() {
            final Deque<T> q = deque;
            final Observer<T> a = downStream;

            while (true) {
                boolean d = done;
                T poll = q.poll();
                boolean empty = poll == null;
                if (checkeTerminated(d, empty, a)) {
                    return;
                }
                if (empty) {
                    downStream.onComplete();
                    break;
                }
                a.onNext(poll);
            }

        }

        private boolean checkeTerminated(boolean d, boolean empty, Observer<T> a) {
            if (over) {
                deque.clear();
                return true;
            }
            if (d) {
                Throwable e = throwable;
                if (e != null) {
                    over = true;
                    a.onError(e);
                    return true;
                } else if (empty) {
                    over = true;
                    a.onComplete();
                    return true;
                }

            }
            return false;
        }
    }


}
